<?php
/**
 * Created by PhpStorm.
 * @author WYZ <wyz@jungo.com.cn>
 * @copyright 深圳市俊网网络有限公司
 */

namespace Selibra\MQ;

use AMQPChannel;
use AMQPConnection;
use AMQPEnvelope;
use AMQPExchange;
use AMQPQueue;
use Selibra\Config\Config;
use Selibra\Di\Annotations\ServerAfterStartup;
use Selibra\Di\DI;
use Swoole\Process;

/**
 * Class ConsumerStarter
 * @package Selibra\MQ
 */
class ConsumerStarter
{

    /**
     * @ServerAfterStartup()
     */
    public function run()
    {
        $consumers = ConsumerCollector::getConsumers();

        foreach ($consumers as $consumer) {
            $process = new Process(function () use ($consumer) {
                go(function () use ($consumer) {
                    $config = Config::get('rabbitmq.connect');
                    $arr = [
                        'host' => $config['host'],
                        'port' => $config['port'],
                        'user' => $config['user'],
                        'password' => $config['password'],
                        'vhost' => $config['vhost'],
                    ];
                    //构造函数   AMQPConnection
                    $con = new AMQPConnection($arr);
                    if (!$con->connect()) {
                        var_dump('连接失败1');
                    }

                    //先通道声明--传入连接的套接字--构造函数 通过通道连接创建消息通道
                    $channel = new AMQPChannel($con);

                    //交换机声明--传入声明的通道-- 构造函数 通过通道连接交换机
                    $exchange = new AMQPExchange($channel);

                    //设置交换机名
                    $exchange->setName($consumer['tags']);//设置通道名称

                    //设置连接方式--直连 [直连，主题，广播]
                    $exchange->setType(AMQP_EX_TYPE_DIRECT);

                    //消息持久化
                    $exchange->setFlags(AMQP_DURABLE);

                    //声明
                    $exchange->declareExchange();

                    //声明队列，绑定交换机和路由
                    $queue = new AMQPQueue($channel);

                    //设置队列名字
                    $queue->setName($consumer['tags']);

                    //消息持久化
                    $queue->setFlags(AMQP_DURABLE);

                    //声明
                    $queue->declareQueue();

                    //绑定获取数据 参数一：交换机名  参数二：路由
                    $queue->bind($consumer['tags'], $consumer['tags']);

                    //消费，没有数据时，阻塞监听获取数据
                    $queue->consume(function (AMQPEnvelope $event, AMQPQueue $queue) use ($consumer) {
                        $body = $event->getBody();
                        $event->getConsumerTag();
                        $class = DI::getObjectContext($consumer['class']);
                        $method = $consumer['method'];
                        $result = $class->$method($body);
                        if ($result !== false) {
                            $state = $queue->ack($event->getDeliveryTag());
                        }
                    });
                });
            });
            $process->start();
        }
        foreach ($consumers as $consumer) {
            Process::wait(true);
        }

    }

}